package com.luych.toolbox.data.service.job;

import com.luych.toolbox.common.enums.*;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketEventPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketLetterPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketNoticePublisher;
import com.luych.toolbox.common.feign.intf.base.service.TaskFeign;
import com.luych.toolbox.common.feign.intf.base.service.TaskLogFeign;
import com.luych.toolbox.common.feign.param.base.service.TaskLogParam;
import com.luych.toolbox.common.feign.param.base.service.TaskParam;
import com.luych.toolbox.common.feign.param.gateway.center.WebSocketEvent;
import com.luych.toolbox.common.feign.view.base.service.TaskLogView;
import com.luych.toolbox.common.feign.view.base.service.TaskView;
import com.luych.toolbox.data.service.constant.Constant;
import com.luych.toolbox.data.service.core.CoreBuilder;
import com.luych.toolbox.data.service.core.CoreConnection;
import com.luych.toolbox.data.service.core.CoreDatabase;
import com.luych.toolbox.data.service.core.CoreTable;
import com.luych.toolbox.data.service.dao.DataSourceDao;
import com.luych.toolbox.data.service.dao.DataTransporterHistoryDao;
import com.luych.toolbox.data.service.dao.DataTransporterHistoryTableDao;
import com.luych.toolbox.data.service.entity.DataSource;
import com.luych.toolbox.data.service.entity.DataTransporterHistory;
import com.luych.toolbox.data.service.entity.DataTransporterHistoryTable;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.text.SimpleDateFormat;
import java.util.*;

public class DataTransporterHistoryExecuteJob extends QuartzJobBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataTransporterHistoryExecuteJob.class);

    private final WebSocketLetterPublisher letterPublisher;
    private final WebSocketNoticePublisher noticePublisher;
    private final WebSocketEventPublisher eventPublisher;

    private final DataSourceDao dataSourceDao;
    private final DataTransporterHistoryDao dataTransporterHistoryDao;
    private final DataTransporterHistoryTableDao dataTransporterHistoryTableDao;

    private final TaskFeign taskFeign;
    private final TaskLogFeign taskLogFeign;

    public DataTransporterHistoryExecuteJob(BusProperties busProperties,
                                            ApplicationEventPublisher applicationEventPublisher,
                                            DataTransporterHistoryDao dataTransporterHistoryDao,
                                            DataSourceDao dataSourceDao,
                                            DataTransporterHistoryTableDao dataTransporterHistoryTableDao,
                                            TaskFeign taskFeign, TaskLogFeign taskLogFeign) {
        this.letterPublisher = new WebSocketLetterPublisher(busProperties, applicationEventPublisher);
        this.noticePublisher = new WebSocketNoticePublisher(busProperties, applicationEventPublisher);
        this.eventPublisher = new WebSocketEventPublisher(busProperties, applicationEventPublisher);

        this.dataTransporterHistoryDao = dataTransporterHistoryDao;
        this.dataSourceDao = dataSourceDao;
        this.dataTransporterHistoryTableDao = dataTransporterHistoryTableDao;
        this.taskFeign = taskFeign;
        this.taskLogFeign = taskLogFeign;
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        String id = context.getJobDetail().getJobDataMap().getString("id");

        DataTransporterHistory transporterHistory = dataTransporterHistoryDao.findById(id).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_TRANSPORTER_HISTORY_NOT_FOUND));
        DataSource dataSourceFrom = dataSourceDao.findById(transporterHistory.getDataSourceFromId()).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_SOURCE_NOT_FOUND));
        DataSource dataSourceTo = dataSourceDao.findById(transporterHistory.getDataSourceToId()).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_SOURCE_NOT_FOUND));

        this.startTransporterHistory(transporterHistory);

        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_TRANSPORTER_RUNNING, transporterHistory.getId()));
        TaskView taskResult = this.createTask(transporterHistory, dataSourceFrom, dataSourceTo);

        LOGGER.info("start to transporter data source [{}] to data source [{}]...", dataSourceFrom.getUrl(), dataSourceTo.getUrl());
        try {
            this.doTransporter(transporterHistory, taskResult, dataSourceFrom, dataSourceTo);
            LOGGER.info("success to save transporter result of data source [{}] to data source [{}]", dataSourceFrom.getUrl(), dataSourceTo.getUrl());
            this.onSuccess(transporterHistory, taskResult);
        } catch (Exception e) {
            LOGGER.info("error to transporter data source [{}] to data source [{}]: {}", dataSourceFrom.getUrl(), dataSourceTo.getUrl(), e.getMessage(), e);
            this.onError(transporterHistory, taskResult, e);
        }
    }

    private void startTransporterHistory(DataTransporterHistory transporterHistory) {
        transporterHistory.setStartDate(new Date());
        transporterHistory.setStatus(DataTransporterHistoryStatus.RUNNING);
        dataTransporterHistoryDao.save(transporterHistory);
    }

    private TaskView createTask(DataTransporterHistory transporterHistory, DataSource dataSourceOne, DataSource dataSourceTwo) {
        TaskParam taskParam = new TaskParam();
        taskParam.setType(TaskType.DATA_TRANSPORTER);
        taskParam.setStatus(TaskStatus.PENDING);
        taskParam.setName("同步数据源");
        taskParam.setDescription("同步数据源[" + dataSourceOne.getName() + "-->" + dataSourceTwo.getName() + "]");
        taskParam.setOwnerId(transporterHistory.getOwnerId());
        TaskView taskResult = taskFeign.create(taskParam);
        taskFeign.start(taskResult.getId());
        return taskResult;
    }

    private TaskLogView createTaskLog(String taskId, String log) {
        LOGGER.info(log);
        TaskLogParam taskLogParam = new TaskLogParam();
        taskLogParam.setTaskId(taskId);
        taskLogParam.setLog(log);
        return taskLogFeign.create(taskLogParam);
    }

    private void doTransporter(DataTransporterHistory transporterHistory, TaskView task, DataSource dataSourceFrom, DataSource dataSourceTo) throws JobExecutionException {
        CoreConnection dataConnectionFrom = CoreBuilder.buildConnection(dataSourceFrom);
        CoreConnection dataConnectionTo = CoreBuilder.buildConnection(dataSourceTo);

        this.createTaskLog(task.getId(), "start get table map ...");
        Map<CoreTable, CoreTable> tableMap = this.getTableMap(dataConnectionFrom, dataConnectionTo, transporterHistory.getTableMap());

        for (Map.Entry<CoreTable, CoreTable> tableEntry : tableMap.entrySet()) {

            DataTransporterHistoryTable transporterTable = new DataTransporterHistoryTable();
            transporterTable.setId(UUID.randomUUID().toString());
            transporterTable.setDataTransporterHistoryId(transporterHistory.getId());
            transporterTable.setTableFrom(tableEntry.getKey().getName());
            transporterTable.setTableTo(tableEntry.getValue().getName());
            transporterTable.setCountFrom(this.getTableCount(dataConnectionFrom, tableEntry.getKey()));
            transporterTable.setCountTo(this.getTableCount(dataConnectionTo, tableEntry.getValue()));
            transporterTable.setCountTransporter(0);
            dataTransporterHistoryTableDao.save(transporterTable);

            this.doTableTransporter(transporterTable, task, dataConnectionFrom, dataConnectionTo, tableEntry.getKey(), tableEntry.getValue());
        }
    }

    private long getTableCount(CoreConnection connection, CoreTable table) {
        List<Map<String, Object>> list = connection.query(table.getCountSql(CaseType.UNCHANGE));
        return list.stream().findFirst().orElse(new HashMap<>()).values().stream().findFirst().map(v -> Long.parseLong(v.toString())).orElse(0L);
    }

    private Map<CoreTable, CoreTable> getTableMap(CoreConnection dataConnectionFrom, CoreConnection dataConnectionTo, Map<String, String> tableMap) throws JobExecutionException {
        CoreDatabase databaseFrom = dataConnectionFrom.getDatabase();
        CoreDatabase databaseTo = dataConnectionTo.getDatabase();

        Map<CoreTable, CoreTable> coreTableMap = new TreeMap<>(Comparator.comparing(CoreTable::getName));
        if (tableMap.isEmpty()) {
            for (CoreTable tableFrom : databaseFrom.getTables()) {
                CoreTable tableTo = databaseTo.getTables().stream()
                        .filter(table -> table.getName().equalsIgnoreCase(tableFrom.getName())).findFirst()
                        .orElseThrow(() -> new JobExecutionException(
                                String.format(Constant.DATA_TRANSPORTER_HISTORY_EXECUTE_ERROR_TO_TABLE_NOT_FOUND, tableFrom.getName())
                        ));
                coreTableMap.put(tableFrom, tableTo);
            }
        } else {
            for (Map.Entry<String, String> tableEntry : tableMap.entrySet()) {
                CoreTable tableFrom = databaseFrom.getTables().stream()
                        .filter(table -> table.getName().equals(tableEntry.getKey())).findFirst()
                        .orElseThrow(() -> new JobExecutionException(
                                String.format(Constant.DATA_TRANSPORTER_HISTORY_EXECUTE_ERROR_FROM_TABLE_NOT_FOUND, tableEntry.getKey())
                        ));
                CoreTable tableTo = databaseTo.getTables().stream()
                        .filter(table -> table.getName().equals(tableEntry.getValue())).findFirst()
                        .orElseThrow(() -> new JobExecutionException(
                                String.format(Constant.DATA_TRANSPORTER_HISTORY_EXECUTE_ERROR_TO_TABLE_NOT_FOUND, tableEntry.getValue())
                        ));
                coreTableMap.put(tableFrom, tableTo);
            }
        }
        return coreTableMap;
    }

    private void startTableTransporter(DataTransporterHistoryTable transporterTable) {
        transporterTable.setStatus(DataTransporterHistoryTableStatus.RUNNING);
        transporterTable.setCountTransporter(0);
        dataTransporterHistoryTableDao.save(transporterTable);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_TRANSPORTER_TABLE_RUNNING, transporterTable.getDataTransporterHistoryId()));
    }

    private void increaseTableTransporterCount(DataTransporterHistoryTable transporterTable, long count) {
        transporterTable.setCountTransporter(count);
        dataTransporterHistoryTableDao.save(transporterTable);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_TRANSPORTER_TABLE_INCREASE, transporterTable.getDataTransporterHistoryId()));
    }

    private void successTableTransporter(DataTransporterHistoryTable transporterTable, long count) {
        transporterTable.setStatus(DataTransporterHistoryTableStatus.SUCCESS);
        transporterTable.setCountTransporter(count);
        dataTransporterHistoryTableDao.save(transporterTable);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_TRANSPORTER_TABLE_SUCCESS, transporterTable.getDataTransporterHistoryId()));
    }

    private void errorTableTransporter(DataTransporterHistoryTable transporterTable, long count) {
        transporterTable.setStatus(DataTransporterHistoryTableStatus.FAILED);
        transporterTable.setCountTransporter(count);
        dataTransporterHistoryTableDao.save(transporterTable);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_TRANSPORTER_TABLE_ERROR, transporterTable.getDataTransporterHistoryId()));
    }

    private Long doTableTransporter(DataTransporterHistoryTable transporterTable, TaskView task,
                                    CoreConnection dataConnectionFrom, CoreConnection dataConnectionTo,
                                    CoreTable tableFrom, CoreTable tableTo) {

        long countTransporter = 0;
        this.startTableTransporter(transporterTable);
        try {
            long logMod = this.getLogMod(transporterTable.getCountFrom());
            this.createTaskLog(task.getId(), "record of from table " + tableFrom.getName() + ": " + transporterTable.getCountFrom() + ", log mod: " + logMod);

            dataConnectionTo.execute(tableTo.getTruncateSql(CaseType.UNCHANGE));
            this.createTaskLog(task.getId(), "to table " + tableFrom.getName() + " has bean truncated");

            long pageSize = 100, pageNo = 1;
            while (pageSize * (pageNo - 1) < transporterTable.getCountFrom()) {
                List<Map<String, Object>> lines = dataConnectionFrom.query(tableFrom.getPageSql(CaseType.UNCHANGE, pageSize, pageNo));
                for (Map<String, Object> line : lines) {
                    dataConnectionTo.execute(tableTo.getInsertSql(CaseType.UNCHANGE), new TreeMap<>(line).values());
                    countTransporter++;
                    if (countTransporter % logMod == 0) {
                        this.increaseTableTransporterCount(transporterTable, countTransporter);
                        this.createTaskLog(task.getId(), "transporter table from " + tableFrom.getName() + " to " + tableTo.getName() + ": " + countTransporter + "/" + transporterTable.getCountFrom());
                    }
                }
                pageNo++;
            }

            this.successTableTransporter(transporterTable, countTransporter);
            this.createTaskLog(task.getId(), "transporter table from " + tableFrom.getName() + " to " + tableTo.getName() + ": " + countTransporter + "/" + transporterTable.getCountFrom());

            return countTransporter;
        } catch (Exception e) {
            this.errorTableTransporter(transporterTable, countTransporter);
            this.createTaskLog(task.getId(), "transporter table from " + tableFrom.getName() + " to " + tableTo.getName() + " error: " + e.getMessage() + ". transported: " + countTransporter + "/" + transporterTable.getCountFrom());

            throw e;
        }
    }

    private long getLogMod(long size) {
        if (size <= 10) return 1;
        if (size <= 100) return 10;
        return (long) Math.pow(10, ((int) Math.log10(size)) - 1);
    }

    private void onSuccess(DataTransporterHistory transporterHistory, TaskView taskResult) {
        String datetime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date());

        transporterHistory.setStatus(DataTransporterHistoryStatus.SUCCESS);
        transporterHistory.setCompleteDate(new Date());
        dataTransporterHistoryDao.save(transporterHistory);

        String content = "数据同步任务[" + transporterHistory.getName() + "]于" + datetime + "执行成功！";
        letterPublisher.success(transporterHistory.getOwnerId(), "数据同步任务执行通知", content);
        noticePublisher.success(transporterHistory.getOwnerId(), content);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_TRANSPORTER_SUCCESS, transporterHistory.getId()));

        taskFeign.complete(taskResult.getId(), TaskStatus.SUCCESS, "成功");
    }

    private void onError(DataTransporterHistory transporterHistory, TaskView taskResult, Exception e) {
        String datetime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date());

        StringWriter writer = new StringWriter();
        e.printStackTrace(new PrintWriter(writer));

        transporterHistory.setStatus(DataTransporterHistoryStatus.FAILED);
        transporterHistory.setErrorMessage(e.getMessage());
        transporterHistory.setErrorTrace(writer.toString());
        dataTransporterHistoryDao.save(transporterHistory);

        String content = "数据同步任务[" + transporterHistory.getName() + "]于" + datetime + "执行异常，请及时处理。";
        noticePublisher.error(transporterHistory.getOwnerId(), content);
        content += "\n\n\n错误信息：\n" + e.getMessage() + "\n\n\n异常堆栈：\n" + writer;
        letterPublisher.error(transporterHistory.getOwnerId(), "数据同步任务执行通知", content);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_TRANSPORTER_ERROR, transporterHistory.getId()));

        taskFeign.complete(taskResult.getId(), TaskStatus.ERROR, e.getMessage());
    }
}
